{
 "metadata": {
  "signature": "sha256:cdc2d0d3e038fc8f876b5effc81e2769cd2662b57456f9e1bb8962a55663b419"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Data Acquisition with NIDAQmx\n",
      "======================================================================\n",
      "\n",
      "These are quick examples of using\n",
      "[ctypes](http://docs.python.org/lib/module-ctypes.html) and numpy to do\n",
      "data acquisition and playback using [National\n",
      "Instrument's](http://ni.com)\n",
      "[NI-DAQmx](http://www.ni.com/dataacquisition/nidaqmx.htm) library. This\n",
      "library allows access to their wide range of data acquisition devices.\n",
      "By using ctypes, we bypass the need for a C compiler. The code below\n",
      "assumes a Windows platform. NI-DAQmx is also available for Linux, but\n",
      "the code below would require a few minor changes, namely loading the\n",
      "shared library and setting the function signatures.\n",
      "\n",
      "See also [Data acquisition with PyUniversalLibrary](Data Acquisition with PyUL).\n",
      "\n",
      "See also projects that wrap NI-DAQmx library with Linux support:\n",
      "[pylibnidaqmx](http://code.google.com/p/pylibnidaqmx/),\n",
      "[pydaqmx](http://code.google.com/p/pydaqmx/),\n",
      "[daqmxbase-swig](http://code.google.com/p/daqmxbase-swig/).\n",
      "\n",
      "OK, enough talk, let's see the code!\n",
      "\n",
      "Analog Acquisition\n",
      "------------------"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "#!python numbers=disable\n",
      "#Acq_IncClk.py\n",
      "# This is a near-verbatim translation of the example program\n",
      "# C:\\Program Files\\National Instruments\\NI-DAQ\\Examples\\DAQmx ANSI C\\Analog In\\Measure Voltage\\Acq-Int Clk\\Acq-IntClk.c\n",
      "import ctypes\n",
      "import numpy\n",
      "nidaq = ctypes.windll.nicaiu # load the DLL\n",
      "##############################\n",
      "# Setup some typedefs and constants\n",
      "# to correspond with values in\n",
      "# C:\\Program Files\\National Instruments\\NI-DAQ\\DAQmx ANSI C Dev\\include\\NIDAQmx.h\n",
      "# the typedefs\n",
      "int32 = ctypes.c_long\n",
      "uInt32 = ctypes.c_ulong\n",
      "uInt64 = ctypes.c_ulonglong\n",
      "float64 = ctypes.c_double\n",
      "TaskHandle = uInt32\n",
      "# the constants\n",
      "DAQmx_Val_Cfg_Default = int32(-1)\n",
      "DAQmx_Val_Volts = 10348\n",
      "DAQmx_Val_Rising = 10280\n",
      "DAQmx_Val_FiniteSamps = 10178\n",
      "DAQmx_Val_GroupByChannel = 0\n",
      "##############################\n",
      "def CHK(err):\n",
      "    \"\"\"a simple error checking routine\"\"\"\n",
      "    if err < 0:\n",
      "        buf_size = 100\n",
      "        buf = ctypes.create_string_buffer('\\000' * buf_size)\n",
      "        nidaq.DAQmxGetErrorString(err,ctypes.byref(buf),buf_size)\n",
      "        raise RuntimeError('nidaq call failed with error %d: %s'%(err,repr(buf.value)))\n",
      "# initialize variables\n",
      "taskHandle = TaskHandle(0)\n",
      "max_num_samples = 1000\n",
      "data = numpy.zeros((max_num_samples,),dtype=numpy.float64)\n",
      "# now, on with the program\n",
      "CHK(nidaq.DAQmxCreateTask(\"\",ctypes.byref(taskHandle)))\n",
      "CHK(nidaq.DAQmxCreateAIVoltageChan(taskHandle,\"Dev1/ai0\",\"\",\n",
      "                                   DAQmx_Val_Cfg_Default,\n",
      "                                   float64(-10.0),float64(10.0),\n",
      "                                   DAQmx_Val_Volts,None))\n",
      "CHK(nidaq.DAQmxCfgSampClkTiming(taskHandle,\"\",float64(10000.0),\n",
      "                                DAQmx_Val_Rising,DAQmx_Val_FiniteSamps,\n",
      "                                uInt64(max_num_samples)));\n",
      "CHK(nidaq.DAQmxStartTask(taskHandle))\n",
      "read = int32()\n",
      "CHK(nidaq.DAQmxReadAnalogF64(taskHandle,max_num_samples,float64(10.0),\n",
      "                             DAQmx_Val_GroupByChannel,data.ctypes.data,\n",
      "                             max_num_samples,ctypes.byref(read),None))\n",
      "print \"Acquired %d points\"%(read.value)\n",
      "if taskHandle.value != 0:\n",
      "    nidaq.DAQmxStopTask(taskHandle)\n",
      "    nidaq.DAQmxClearTask(taskHandle)\n",
      "print \"End of program, press Enter key to quit\"\n",
      "raw_input()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Analog Generation\n",
      "-----------------"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "#!python numbers=disable\n",
      "\"\"\"\n",
      "This is an interpretation of the example program\n",
      "C:\\Program Files\\National Instruments\\NI-DAQ\\Examples\\DAQmx ANSI C\\Analog Out\\Generate Voltage\\Cont Gen Volt Wfm-Int Clk\\ContGen-IntClk.c\n",
      "This routine will play an arbitrary-length waveform file.\n",
      "This module depends on:\n",
      "numpy\n",
      "Adapted by Martin Bures [ mbures { @ } zoll { . } com ]\n",
      "\"\"\"\n",
      "# import system libraries\n",
      "import ctypes\n",
      "import numpy\n",
      "import threading\n",
      "# load any DLLs\n",
      "nidaq = ctypes.windll.nicaiu # load the DLL\n",
      "##############################\n",
      "# Setup some typedefs and constants\n",
      "# to correspond with values in\n",
      "# C:\\Program Files\\National Instruments\\NI-DAQ\\DAQmx ANSI C Dev\\include\\NIDAQmx.h\n",
      "# the typedefs\n",
      "int32 = ctypes.c_long\n",
      "uInt32 = ctypes.c_ulong\n",
      "uInt64 = ctypes.c_ulonglong\n",
      "float64 = ctypes.c_double\n",
      "TaskHandle = uInt32\n",
      "# the constants\n",
      "DAQmx_Val_Cfg_Default = int32(-1)\n",
      "DAQmx_Val_Volts = 10348\n",
      "DAQmx_Val_Rising = 10280\n",
      "DAQmx_Val_FiniteSamps = 10178\n",
      "DAQmx_Val_ContSamps = 10123\n",
      "DAQmx_Val_GroupByChannel = 0\n",
      "##############################\n",
      "class WaveformThread( threading.Thread ):\n",
      "    \"\"\"\n",
      "    This class performs the necessary initialization of the DAQ hardware and\n",
      "    spawns a thread to handle playback of the signal.\n",
      "    It takes as input arguments the waveform to play and the sample rate at which\n",
      "    to play it.\n",
      "    This will play an arbitrary-length waveform file.\n",
      "    \"\"\"\n",
      "    def __init__( self, waveform, sampleRate ):\n",
      "        self.running = True\n",
      "        self.sampleRate = sampleRate\n",
      "        self.periodLength = len( waveform )\n",
      "        self.taskHandle = TaskHandle( 0 )\n",
      "        self.data = numpy.zeros( ( self.periodLength, ), dtype=numpy.float64 )\n",
      "        # convert waveform to a numpy array\n",
      "        for i in range( self.periodLength ):\n",
      "            self.data[ i ] = waveform[ i ]\n",
      "        # setup the DAQ hardware\n",
      "        self.CHK(nidaq.DAQmxCreateTask(\"\",\n",
      "                          ctypes.byref( self.taskHandle )))\n",
      "        self.CHK(nidaq.DAQmxCreateAOVoltageChan( self.taskHandle,\n",
      "                                   \"Dev1/ao0\",\n",
      "                                   \"\",\n",
      "                                   float64(-10.0),\n",
      "                                   float64(10.0),\n",
      "                                   DAQmx_Val_Volts,\n",
      "                                   None))\n",
      "        self.CHK(nidaq.DAQmxCfgSampClkTiming( self.taskHandle,\n",
      "                                \"\",\n",
      "                                float64(self.sampleRate),\n",
      "                                DAQmx_Val_Rising,\n",
      "                                DAQmx_Val_FiniteSamps,\n",
      "                                uInt64(self.periodLength)));\n",
      "        self.CHK(nidaq.DAQmxWriteAnalogF64( self.taskHandle,\n",
      "                              int32(self.periodLength),\n",
      "                              0,\n",
      "                              float64(-1),\n",
      "                              DAQmx_Val_GroupByChannel,\n",
      "                              self.data.ctypes.data,\n",
      "                              None,\n",
      "                              None))\n",
      "        threading.Thread.__init__( self )\n",
      "    def CHK( self, err ):\n",
      "        \"\"\"a simple error checking routine\"\"\"\n",
      "        if err < 0:\n",
      "            buf_size = 100\n",
      "            buf = ctypes.create_string_buffer('\\000' * buf_size)\n",
      "            nidaq.DAQmxGetErrorString(err,ctypes.byref(buf),buf_size)\n",
      "            raise RuntimeError('nidaq call failed with error %d: %s'%(err,repr(buf.value)))\n",
      "        if err > 0:\n",
      "            buf_size = 100\n",
      "            buf = ctypes.create_string_buffer('\\000' * buf_size)\n",
      "            nidaq.DAQmxGetErrorString(err,ctypes.byref(buf),buf_size)\n",
      "            raise RuntimeError('nidaq generated warning %d: %s'%(err,repr(buf.value)))\n",
      "    def run( self ):\n",
      "        counter = 0\n",
      "        self.CHK(nidaq.DAQmxStartTask( self.taskHandle ))\n",
      "    def stop( self ):\n",
      "        self.running = False\n",
      "        nidaq.DAQmxStopTask( self.taskHandle )\n",
      "        nidaq.DAQmxClearTask( self.taskHandle )\n",
      "if __name__ == '__main__':\n",
      "    import time\n",
      "    # generate a time signal 5 seconds long with 250Hz sample rate\n",
      "    t = numpy.arange( 0, 5, 1.0/250.0 )\n",
      "    # generate sine wave\n",
      "    x = sin( t )\n",
      "    mythread = WaveformThread( x, 250 )\n",
      "    # start playing waveform\n",
      "    mythread.start()\n",
      "    # wait 5 seconds then stop\n",
      "    time.sleep( 5 )\n",
      "    mythread.stop()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    }
   ],
   "metadata": {}
  }
 ]
}